package com.icehand.canal.client;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.google.common.base.Strings;
import com.google.common.collect.Lists;
import com.icehand.canal.client.transfer.TransponderFactory;
import com.icehand.canal.config.CanalConfig;

import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.util.List;
import java.util.Map;
import java.util.Objects;

/**
 * @author icehand
 */
public abstract class BaseCanalClient implements CanalClient{
    private volatile boolean running;

    private CanalConfig canalConfig;

    protected final TransponderFactory factory;

    public BaseCanalClient(CanalConfig config,TransponderFactory factory){
        Objects.requireNonNull(config,"canal config can not be null");
        Objects.requireNonNull(factory,"transponder factory can not be null");
        if(null == config.getInstances() || config.getInstances().size() <1) {
            throw new CanalClientException("canal config is null");
        }
        this.canalConfig = config;
        this.factory = factory;
    }

    protected abstract void process(CanalConnector connector, CanalConfig.Instance instance, String key);

    @Override
    public void start() {
        Map<String,CanalConfig.Instance> instanceMap = canalConfig.getInstances();
        instanceMap.forEach((k,v)-> process(createCanalConnector(k,v),v,k));
    }

    private CanalConnector createCanalConnector(String key, CanalConfig.Instance instance){
        CanalConnector canalConnector;
        if(instance.isClusterEnabled()) {
            List<SocketAddress> addressList = Lists.newArrayList();
            instance.getZookeeperAddress().forEach(s -> {
                String[] arr = s.split(":");
                if(arr.length != 2){
                    throw new CanalClientException("error parsing zookeeper address: {}"+s);
                }
                addressList.add(new InetSocketAddress(arr[0],Integer.parseInt(arr[1])));
            });
            canalConnector = CanalConnectors.newClusterConnector(addressList, key,
                    instance.getUserName(),
                    instance.getPassword());
        } else {
            canalConnector = CanalConnectors.newSingleConnector(new InetSocketAddress(instance.getHost(), instance.getPort()),
                    key,
                    instance.getUserName(),
                    instance.getPassword());
        }

        canalConnector.connect();
        if(!Strings.isNullOrEmpty(instance.getFilter())){
            canalConnector.subscribe(instance.getFilter());
        }else {
            canalConnector.subscribe();
        }

        canalConnector.rollback();
        return canalConnector;
    }


    @Override
    public void stop() {
        this.running = false;
    }

    @Override
    public boolean isRunning() {
        return this.running;
    }
}
